package com.sz.ly.zt.schedule.master;

import com.alibaba.fastjson.JSON;
import com.sz.ly.zt.schedule.core.Conf;
import com.sz.ly.zt.schedule.core.MasterServer;
import com.sz.ly.zt.schedule.core.ScheduleManager;
import com.sz.ly.zt.schedule.core.ZTScheduleJob;
import com.sz.ly.zt.schedule.schedule.QuartzManager;
import com.sz.ly.zt.schedule.schedule.TaskDistributor;
import com.sz.ly.zt.schedule.task.ExecuteLogClearTask;
import com.sz.ly.zt.schedule.task.ExecuteWorkerClearTask;
import com.sz.ly.zt.schedule.task.MasterHeartBeatTask;
import com.sz.ly.zt.schedule.task.WorkerHeartBeatTask;
import com.sz.ly.zt.schedule.worker.ZTScheduleWorker;
import com.sz.ly.zt.schedule.zk.CuratorClient;
import com.sz.ly.zt.schedule.zk.ZKTools;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.utils.CloseableUtils;
import org.quartz.CronTrigger;
import org.quartz.JobDetail;
import org.quartz.JobKey;
import org.quartz.TriggerKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import static org.quartz.CronScheduleBuilder.cronSchedule;
import static org.quartz.JobBuilder.newJob;
import static org.quartz.TriggerBuilder.newTrigger;

/**
 * Created by ly-meibaiqi on 2017/3/30.
 * meiboqi@126.com
 */
public class ZTScheduleMaster {
    private static final transient Logger logger = LoggerFactory.getLogger(ZTScheduleWorker.class);

    /**
     * 当前调度服务的信息
     */
    protected MasterServer masterServer;
    ScheduledExecutorService service = Executors
            .newScheduledThreadPool(3);


    public static void main(String args[]) throws Exception {
        final ZTScheduleMaster master = new ZTScheduleMaster();
     /*   if(checkMasterExists()){
            String errorMsg="Master已经启动，不需要再启动！";
            throw new Exception(errorMsg);
        }else{
            master.start();
        }*/
        master.start();

        /**
         * jvm异常退出时执行代码.
         * 1、程序正常退出.
         * 2、使用System.exit().
         * 3、终端使用Ctrl+C触发的中断.
         * 4、系统关闭.
         * 5、OutOfMemory宕机.
         * 6、使用Kill pid命令干掉进程（注：在使用kill -9 pid时，是不会被调用的.
         */
        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
            public void run() {
                master.stop();
            }
        }));
    }

    /**
     * 启动Master.
     * 1：初始化(创建)Master.
     * 2：注册Master到ZK.
     * 3：启动心跳.
     * 4：监听任务路径.
     */
    private void start() {
        masterServer = MasterServer.createMaster();

        /***********注册Master***********/
        try {
            ScheduleManager.registerMasterServer(masterServer);
        } catch (Exception e) {
            logger.error("注册Master失败!,Master信息如下：" + masterServer.toString());
        }

        /***********启动心跳任务***********/
        this.startHeartBeat(masterServer);

        /***************启动监听器**************/
        try {
            this.listenerToTaskCommand(masterServer);
        } catch (Exception e) {
            logger.error(e.getMessage());
        }

        /****************处理其他事项***************/
        this.timedClearExecutionLog();//定时清除执行日志.
        this.timedClearExecutionWorker();//定时清除过时worker
        //service.shutdown();
    }

    /**
     * 启动节点心跳.
     *
     * @param master
     */
    private void startHeartBeat(MasterServer master) {
        MasterHeartBeatTask heartBeatTask = new MasterHeartBeatTask(master);
        service.scheduleAtFixedRate(heartBeatTask, 10, 60, TimeUnit.SECONDS);
    }

    /**
     * 监听task-command目录.
     *
     * @param workerServer
     * @throws Exception
     */
    private static void listenerToTaskCommand(MasterServer workerServer) throws Exception {
        String commandPath = Conf.TASK_COMMAND_PATH;
        final CuratorFramework client = CuratorClient.getCuratorClient();
        client.start();
        final PathChildrenCache commandPathCache = new PathChildrenCache(client, commandPath, true);
        commandPathCache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event) throws Exception {
                // String value = new String(client.getData().forPath(event.getData().getPath()));
                //ZTScheduleJob job = JSON.parseObject(value, ZTScheduleJob.class);
                switch (event.getType()) {
                    case CHILD_ADDED://启动调度.
                        String value = new String(client.getData().forPath(event.getData().getPath()));
                        ZTScheduleJob job = JSON.parseObject(value, ZTScheduleJob.class);
                        QuartzManager.addJob(job);
                        break;
                    case CHILD_UPDATED://更状态新.
                        // QuartzManager.modifyJobTime(job);
                        break;
                    case CHILD_REMOVED://移除调度
                        String removeValue = new String(event.getData().getData());
                        ZTScheduleJob r_job = JSON.parseObject(removeValue, ZTScheduleJob.class);
                        QuartzManager.removeJob(r_job);
                        break;
                    default:
                        break;
                }
            }
        });
        commandPathCache.start();
    }

    /**
     * 定时清除执行日志.
     * 每天00:00:01执行日志清除.
     */
    private void timedClearExecutionLog() {
        ExecuteLogClearTask logClearTask = new ExecuteLogClearTask();
        long oneDay = 24 * 60 * 60 * 1000;
        long initDelay  = getTimeMillis("00:00:01") - System.currentTimeMillis();
        initDelay = initDelay > 0 ? initDelay : oneDay + initDelay;
        service.scheduleAtFixedRate(logClearTask, initDelay, oneDay, TimeUnit.MILLISECONDS);
    }

    /**
     * 定时清除过时worker
     */
    private void timedClearExecutionWorker(){
        ExecuteWorkerClearTask workerClearTask=new ExecuteWorkerClearTask();
        service.scheduleAtFixedRate(workerClearTask, 10, 60, TimeUnit.SECONDS);
    }

    /**
     * 获取指定时间对应的毫秒数
     * @param time "HH:mm:ss"
     * @return
     */
    private static long getTimeMillis(String time) {
        try {
            DateFormat dateFormat = new SimpleDateFormat("yy-MM-dd HH:mm:ss");
            DateFormat dayFormat = new SimpleDateFormat("yy-MM-dd");
            Date curDate = dateFormat.parse(dayFormat.format(new Date()) + " " + time);
            return curDate.getTime();
        } catch (ParseException e) {
            e.printStackTrace();
        }
        return 0;
    }

    /**
     * 判断Master是否已经启动
     * @return
     */
    private static boolean checkMasterExists(){
        final CuratorFramework client = CuratorClient.getCuratorClient();
        client.start();
        boolean result=false;
        try {
           if(ZKTools.exists(client,Conf.MASTER_NODE_PATH ) != null){
               String value = new String(ZKTools.getData(client,Conf.MASTER_NODE_PATH));
               MasterServer server = JSON.parseObject(value,MasterServer.class);
               long time = server.getHeartBeatTime().getTime();
               if(System.currentTimeMillis()-time <= Conf.WORKER_SURVIVAL_TIME){
                   result=true;
               }
           }
        } catch (Exception e) {
            logger.error(e.getMessage());
        }finally {
            CloseableUtils.closeQuietly(client);
        }
        return result;
    }

    /**
     * Master意外退出.
     */
    public void stop() {
        //TODO master退出时需要处理的事项在该方法实现.
        logger.warn("Master意外退出，正在清理Master残留信息...");
        unRegisterMaster();
    }

    /**
     *Master意外退出时，注销Master.
     */
    private void unRegisterMaster() {
        final CuratorFramework client = CuratorClient.getCuratorClient();
        client.start();
        try {
            ZKTools.deletePath(client,Conf.MASTER_NODE_PATH+"/"+this.masterServer.getIp());
        } catch (Exception e) {
            CloseableUtils.closeQuietly(client);
        }
    }
}


